package com.yuninglong.demo1;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.util.List;

/**
 * @author yuninglong
 */
public class SocketWindowWordCount {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

//        这创建了一个字符串类型的 DataStream。
//       DataStream 是 Flink 中做流处理的核心 API，上面定义了非常多常见的操作（如，过滤、转换、聚合、窗口、关联等）。
//      在本示例中，我们感兴趣的是每个单词在特定时间窗口中出现的次数，比如说5秒窗口。
//     为此，我们首先要将字符串数据解析成单词和次数（使用Tuple2表示），第一个字段是单词，第二个字段是次数，次数初始值都设置成了1。
//    我们实现了一个 flatmap 来做解析的工作，因为一行数据中可能有多个单词
        DataStream text = env.socketTextStream("192.168.202.10", 9000, "\n");
        DataStream<String> wordCounts = text.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {

            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> collector) throws Exception {
                for (String word : value.split("\\s")) {
                    collector.collect(Tuple2.of(word, 1));
                }
            }

        })
//                .returns(Tuple2.class)
                .keyBy(0)
                .timeWindow(Time.seconds(5))
                .sum(1);
        ;

        // 将结果打印到控制台，注意这里使用的是单线程打印，而非多线程
        wordCounts.print().setParallelism(1);

        env.execute("Socket Window WordCount");

    }

}
